﻿Imports System
Imports Microsoft.SqlServer.Dts.Pipeline
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
Imports Microsoft.SqlServer.Dts.Runtime.Wrapper
Imports System.Threading

Public Class RowData
    Public PackageRowId As Integer
    Public Result As Integer
    Public DataList As New List(Of ColumnData)
End Class

Public Class ColumnData
    'Public PackageRowId As Integer

    Public DtsVariable As DTSVariable
    Public ColumnValue As Object
End Class

<DtsPipelineComponent(ComponentType:=ComponentType.Transform, DisplayName:="SSIS Package Execute", UITypeName:="PkgExecUI.PkgExecUI,PkgExecUI,Version=1.0.0.0,Culture=neutral,PublicKeyToken=d0322a67ff51c402")> _
Public Class PkgExec
    Inherits PipelineComponent

    Private outputBuffer As PipelineBuffer
    'Private inputColumnBufferIndexes As Integer()
    'Private outputColumnBufferIndexes As Integer()

    Dim RowList As New List(Of RowData)

    Dim File As String
    Dim MaxDop As Integer
    Dim ColumnMapping As String
    Dim ReadPackage As New SSISPackage
    'Dim Packages() As SSISPackage
    Dim ThrdRowOn As Integer = 0

    Sub New()

    End Sub



    Public Overrides Sub ProvideComponentProperties()
        ' Let the base class add the input and output objects.
        MyBase.ProvideComponentProperties()
        MyBase.RemoveAllInputsOutputsAndCustomProperties()
        ' Name the input and output, and make the
        ' output asynchronous.
        If (ComponentMetaData.InputCollection.Count = 0) Then
            ComponentMetaData.InputCollection.[New]()
            ComponentMetaData.InputCollection(0).Name = "Input"
        End If
        If (ComponentMetaData.OutputCollection.Count = 0) Then
            ComponentMetaData.OutputCollection.[New]()
            ComponentMetaData.OutputCollection(0).Name = "Output"
            ComponentMetaData.OutputCollection(0).SynchronousInputID = 0

            Dim outputColumn As IDTSOutputColumn100 = ComponentMetaData.OutputCollection(0).OutputColumnCollection.[New]
            outputColumn.Name = "PackageRowId"
            outputColumn.SetDataTypeProperties(Microsoft.SqlServer.Dts.Runtime.Wrapper.DataType.DT_I4, 0, 0, 0, 0)

            Dim outputColumn2 As IDTSOutputColumn100 = ComponentMetaData.OutputCollection(0).OutputColumnCollection.[New]
            outputColumn2.Name = "PackageResult"
            outputColumn2.SetDataTypeProperties(Microsoft.SqlServer.Dts.Runtime.Wrapper.DataType.DT_I4, 0, 0, 0, 0)
        End If

        ComponentMetaData.CustomPropertyCollection.[New]()
        ComponentMetaData.CustomPropertyCollection(0).Name = "ColumnMapping"
        ComponentMetaData.CustomPropertyCollection.[New]()
        ComponentMetaData.CustomPropertyCollection(1).Name = "FileName"
        ComponentMetaData.CustomPropertyCollection.[New]()
        ComponentMetaData.CustomPropertyCollection(2).Name = "MaxDop"
    End Sub

    Public Sub ReadCustomProps()
        Dim Prop As Microsoft.SqlServer.Dts.Pipeline.Wrapper.IDTSCustomProperty100 = Nothing
        For Each Prop In ComponentMetaData.CustomPropertyCollection
            If (Prop.Name = "FileName") Then
                File = Prop.Value
            End If
            If (Prop.Name = "MaxDop") Then
                MaxDop = CType(Prop.Value, Integer)
            End If
            If (Prop.Name = "ColumnMapping") Then
                ColumnMapping = Prop.Value

                'Dim MappingAtt() As String
                'MappingAtt = Split(Prop.Value, "|")
                'For Each MappingOn As String In MappingAtt
                '    Dim ThisAtt = Split(MappingOn, ":")

                'Next

            End If

        Next
    End Sub
    Public Overrides Sub PreExecute()

        Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)

        ReadCustomProps()

        ReadPackage.File = File
        ReadPackage.GetVarList()
        ReadPackage.SetColumnMapping(ColumnMapping, input, BufferManager)

    End Sub
    Public Overrides Sub PrimeOutput(ByVal outputs As Integer, ByVal outputIDs As Integer(), ByVal buffers As PipelineBuffer())

        If buffers.Length <> 0 Then
            outputBuffer = buffers(0)
        End If

    End Sub

    Public Sub BuildInternalBuffer(ByRef buffer As Microsoft.SqlServer.Dts.Pipeline.PipelineBuffer)
        Dim NewRow As New RowData
        RowList.Add(NewRow)

        NewRow.PackageRowId = buffer.GetInt32(ReadPackage.PackageRowIdBufferId)
        For Each Col As DTSVariable In ReadPackage.varList

            Dim NewCol As New ColumnData
            NewCol.DtsVariable = Col

            Select Case NewCol.DtsVariable.DataType
                Case TypeCode.String
                    NewCol.ColumnValue = CType(buffer.GetString(Col.BufferId), Object)
                Case TypeCode.Int32
                    NewCol.ColumnValue = CType(buffer.GetInt32(Col.BufferId), Object)
                Case TypeCode.DateTime
                    NewCol.ColumnValue = CType(buffer.GetDateTime(Col.BufferId), Object)


            End Select

            NewRow.DataList.Add(NewCol)

        Next

    End Sub

    Public Overrides Sub ProcessInput(ByVal inputID As Integer, ByVal buffer As PipelineBuffer)
        While (buffer.NextRow())
            BuildInternalBuffer(buffer)
        End While
        ' Advance the buffer to the next row.

        If buffer.EndOfRowset = True Then
            CreateWorkers()
            ProcessRows()
            ' EndOfRowset on the input buffer is true.
            ' Set the end of row set on the output buffer.
            outputBuffer.SetEndOfRowset()
        End If
    End Sub

    Dim ThreadList As New List(Of Thread)

    Public Sub CreateWorkers()
        For x As Integer = 1 To Integer.Parse(MaxDop) - 1
            Dim t As Thread = New Thread(AddressOf Me.BackgroundProcess)

            ThreadList.Add(t)
        Next
    End Sub
    Dim Locker As New Object


    Private Sub BackgroundProcess()
        Dim ThrdPackage As New SSISPackage
        ThrdPackage.File = ReadPackage.File
        ThrdPackage.PackageRowIdBufferId = ReadPackage.PackageRowIdBufferId
        ThrdPackage.GetVarList()


        While (True)
            Dim RowToProcess As RowData
            SyncLock Locker
                If (ThrdRowOn >= RowList.Count) Then
                    Return
                End If
                RowToProcess = RowList(ThrdRowOn)
                ThrdRowOn += 1
            End SyncLock
            ThrdPackage.ExecPackage(RowToProcess)
        End While
    End Sub

    Public Sub ProcessRows()

        For Each Thread As Thread In ThreadList
            Thread.Start()
        Next
        For Each Thread As Thread In ThreadList
            While (Thread.IsAlive())
                System.Threading.Thread.Sleep(100)
            End While
        Next

    End Sub
End Class

